package com.shujia.flink.core

import java.{lang, util}

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

object Demo8ListState {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentsDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
      * 实时统计每个班级的平均年龄
      *
      */

    val kvDS: DataStream[(String, Int)] = studentsDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(4), split(2).toInt)
    })


    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)


    val avgAgeDS: DataStream[(String, Double)] = keyByDS.process(new AvgProcess)

    avgAgeDS.print()


    env.execute()


  }

}

class AvgProcess extends KeyedProcessFunction[String, (String, Int), (String, Double)] {


  var listState: ListState[Int] = _

  override def open(parameters: Configuration): Unit = {
    //定义状态

    val context: RuntimeContext = getRuntimeContext


    //集合状态的描述对象
    val listStateDesc = new ListStateDescriptor[Int]("ages", classOf[Int])

    listState = context.getListState(listStateDesc)


  }

  override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Double)]#Context, out: Collector[(String, Double)]): Unit = {

    //往集合状态中增加一个元素
    listState.add(value._2)


    //获取同一个班级所有的年龄，java中的迭代器
    val iterable: lang.Iterable[Int] = listState.get()

    val ages: util.Iterator[Int] = iterable.iterator()

    //将java集合转换成scala集合
    import scala.collection.JavaConversions._

    val agesList: List[Int] = ages.toList

    //计算平均年龄
    val avgAge: Double = agesList.sum / agesList.size.toDouble


    //将数据发送到下游
    out.collect((value._1, avgAge))
  }


}
